package io.kubesphere.devops.core;

import io.kubesphere.devops.common.ExecutorRule;
import io.kubesphere.devops.common.Rule;
import io.kubesphere.devops.core.struct.ColumnMeta;
import io.kubesphere.devops.core.struct.TableMeta;
import io.kubesphere.devops.core.struct.TableMetaCacheFactory;
import io.kubesphere.devops.model.UserMergingRule;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.*;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * <p> UpdateWithStandloneRule 基于单库修改直接执行update 语句既可
 *
 * @author gaochuan
 * @version 0.1.0
 * @date 2021/11/18 17:26
 * @since 2021/11/18
 */
@Slf4j
@Rule
@Service
public class UpdateWithStandloneRule implements ExecutorRule {
    private final DataSource dynamicRoutingDataSource;
    private final DatasourceConfigMap datasourceConfigMap;

    public UpdateWithStandloneRule(DataSource dynamicRoutingDataSource, DatasourceConfigMap datasourceConfigMap) {
        this.dynamicRoutingDataSource = dynamicRoutingDataSource;
        this.datasourceConfigMap = datasourceConfigMap;
    }

    @Override
    public void process(String sourceIdentificationId, String targetIdentificationId, UserMergingRule rule) throws SQLException {

        //todo 切换到目标sourceIdentificationId 所在数据源（理论上sourceIdentificationId 所在数据源和targetIdentificationId相同）
        //执行sql 即可  ==》update table set phrcode = targetIdentificationId where phrcode=sourceIdentificationId
        int sourceIdDatasourceIndex = PartitionZoneLocator.calculate(StringRemoveBackquote.removeBackquote(sourceIdentificationId));
        int targetIdDatasourceIndex = PartitionZoneLocator.calculate(StringRemoveBackquote.removeBackquote(targetIdentificationId));

        log.info("计算账户所在分片,被合并账户所在分库: {},目标账户所在分库 : {}", sourceIdDatasourceIndex, targetIdDatasourceIndex);
        //假设可以对应sourceIdentificationId index 分库为ZONE2,表所在schema,数据源key = zone2+schema
        DynamicDataSourceContextHolder.routePartition(datasourceConfigMap.getPartitionZone(sourceIdDatasourceIndex), rule);
        //计算被合并数据数量,按需启动执行分片 select count(*) from account where id = sourceIdDatasourceIndex;if count >1000 开启分片模式，每个分片处理数量为1000，持久化分片
        int count = count(sourceIdentificationId, rule);
        if (count == 0) {
            return;
        }
        log.info("查询被合并账户所在分库数据情况,{},开始处理{} 分库数据", count > 1000 ? "处理行数超过1000开启并行处理" : "处理行数小于1000 单线程处理", sourceIdDatasourceIndex);

        for (int i = 0; i <= count / 1000; i++) {
            loop(sourceIdentificationId,
                    targetIdentificationId,
                    rule,
                    sourceIdDatasourceIndex,
                    targetIdDatasourceIndex,
                    i + 1
            );
        }
    }

    List<Map<String, Object>> queryPage(int pageSize, int pageNo, Connection conn, String sourceIdentificationId, UserMergingRule rule, TableMeta tableMeta) throws SQLException {
        //select * from table limit (pageNo-1)*pageSize, pageSize;
        String sql = String.format(
                "select * from %s.%s where %s = ? limit %s, %s",
                rule.getSchemaName(),
                rule.getTableName(),
                rule.getMergeIdentificationName(),
                (pageNo - 1) * pageSize, pageSize
        );
        try (PreparedStatement ps = conn.prepareStatement(sql)) {
            ps.setString(1, sourceIdentificationId);
            try (ResultSet rs = ps.executeQuery()) {
                List<Map<String, Object>> list = convertList(rs);
                return list;
            }
        }

    }

    private void loop(String sourceIdentificationId,
                      String targetIdentificationId,
                      UserMergingRule rule,
                      int sourceIdDatasourceIndex,
                      int targetIdDatasourceIndex,
                      int loopIndex
    ) throws SQLException {

        //todo update sql execute
        //假设可以对应sourceIdentificationId index 分库为ZONE2,表所在schema,数据源key = zone2+schema
        DynamicDataSourceContextHolder.routePartition(datasourceConfigMap.getPartitionZone(sourceIdDatasourceIndex), rule);
        List<Map<String, Object>> records;
        //查询待合并数据,申请连接使用后关闭连接
        try (Connection connection = dynamicRoutingDataSource.getConnection()) {
            TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache("mysql").getTableMeta(
                    connection, rule.getTableName(), rule.getSchemaName());

            records = queryPage(1000, loopIndex, connection, sourceIdentificationId, rule, tableMeta);
            if (null == records || records.size() == 0) {
                return;
            }
            //构造删除对象主键列表
            List<Object> ids = records.stream().map(a -> a.get(rule.getTablePkName())).collect(Collectors.toList());
            //把这批数据给更新了
            updateRecords(connection, rule, ids, tableMeta, targetIdentificationId);
        }
    }

    int updateRecords(Connection conn, UserMergingRule rule, List<Object> ids, TableMeta tableMeta, String targetIdentificationId) throws SQLException {
        Map<String, ColumnMeta> columnMetaMap = tableMeta.getAllColumns();
        String sql = String.format("update %s.%s  set %s = %s where %s in (%s)",
                rule.getSchemaName(),
                rule.getTableName(),
                rule.getMergeIdentificationName(),
                targetIdentificationId,
                rule.getTablePkName(),
                ids.stream().map(m -> "?").collect(Collectors.joining(", "))
        );

        try (PreparedStatement ps = conn.prepareStatement(sql)) {
            int index = 1;
            for (Object id : ids) {
                if (id instanceof Integer) {
                    ps.setInt(index++, (Integer) id);
                }
                if (id instanceof String) {
                    ps.setString(index++, (String) id);
                }
            }
            return ps.executeUpdate();
        }
    }

    int count(String sourceIdentificationId, UserMergingRule rule) throws SQLException {
        String sql = String.format(
                "select count(*) rowCount from %s.%s where %s = ?",
                rule.getSchemaName(),
                rule.getTableName(),
                rule.getMergeIdentificationName()
        );
        //select * from table limit (pageNo-1)*pageSize, pageSize;
        try (Connection connection = dynamicRoutingDataSource.getConnection()) {
            try (PreparedStatement ps = connection.prepareStatement(sql)) {
                ps.setString(1, sourceIdentificationId);
                try (ResultSet rs = ps.executeQuery()) {
                    while (rs.next()) {
                        return rs.getInt("rowCount");
                    }
                }
            }
        }
        return 0;

    }

    public List<Map<String, Object>> convertList(ResultSet rs) {
        List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
        try {
            ResultSetMetaData md = rs.getMetaData();
            int columnCount = md.getColumnCount();
            while (rs.next()) {
                Map<String, Object> rowData = new HashMap<String, Object>();
                for (int i = 1; i <= columnCount; i++) {
                    rowData.put(md.getColumnName(i), rs.getObject(i));
                }
                list.add(rowData);
            }
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } finally {
            try {
                if (rs != null)
                    rs.close();
                rs = null;
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return list;
    }
}
